package com.peng.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class TwoConsumer extends Thread{
    public static void main(String[] args) {
        TwoConsumer consumer1=new TwoConsumer("consumer-1");
        consumer1.start();
        TwoConsumer consumer2=new TwoConsumer("consumer-2");
        consumer2.start();
    }
    private String consumerName=null;
    public TwoConsumer(String consumerName){
        this.consumerName=consumerName;
    }

    @Override
    public void run() {
        System.out.println("消费者1"+consumerName+"正在消费消息");
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");//Kafka集群
        props.put("group.id", "test");//消费者组id
        props.put("enable.auto.commit", "true");//自动提交 ，提交offset
        props.put("auto.commit.interval.ms", "1000");//提交延时1s
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(props);
        //3 指定topic
        consumer.subscribe(Arrays.asList("first","first-topic","second"));
        while(true){
            ConsumerRecords<String,String> consumerRecords=consumer.poll(100);
            for(ConsumerRecord<String,String> record:consumerRecords){
                System.out.println(record.topic()+"--"+record.partition()+"--"+record.value());
                try{
                    sleep(3000);
                }catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
